package test;

import java.util.HashMap;

import org.apache.spark.api.java.function.Function;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaReceiverInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.twitter.TwitterUtils;

import twitter4j.Status;
import twitter4j.auth.Authorization;
import twitter4j.auth.OAuthAuthorization;
import twitter4j.conf.Configuration;
import twitter4j.conf.ConfigurationBuilder;

public final class JavaQueueStream {
	private JavaQueueStream() {
	}

	public static void main(String[] args) throws Exception {

		Configuration conf = new ConfigurationBuilder()
				.setOAuthAccessToken(
						"2255068147-v32adeOq6b6wzM3KvnjusTHhuEykNxgkZ3PScyR")
				.setOAuthAccessTokenSecret(
						"vXkR9jtECj0CSPpH9lEyfsE8v8citcZ5N61VaqhKJll3n")
				.setOAuthConsumerKey("pcGABkK2IcOgImFtS20g")
				.setOAuthConsumerSecret(
						"v9KL2VMCvYslz5M7MLAu9DkVa3dcAEwNuSwOs5xU").build();
		Authorization auth = new OAuthAuthorization(conf);

		JavaStreamingContext ssc = new JavaStreamingContext(
				"spark://DFX1JTS1:6951",
				"Tutorial2",
				new Duration(1000),
				"C:\\cygdrive\\c\\blaze\\installs\\spark-1.0.2-bin-hadoop1",
				new String[] {
						"C:\\Users\\nipunkumar\\Desktop\\kalptest\\test\\target\\test-0.0.1-SNAPSHOT-tests.jar",
						"C:\\Users\\nipunkumar\\.m2\\repository\\org\\twitter4j\\twitter4j-stream\\3.0.3\\twitter4j-stream-3.0.3.jar",
						"C:\\Users\\nipunkumar\\.m2\\repository\\org\\twitter4j\\twitter4j-core\\3.0.3\\twitter4j-core-3.0.3.jar",
						"C:\\Users\\nipunkumar\\.m2\\repository\\org\\apache\\spark\\spark-streaming-twitter_2.10\\1.0.2\\spark-streaming-twitter_2.10-1.0.2.jar"},
				new HashMap<String, String>());
		JavaReceiverInputDStream<Status> tweets = TwitterUtils.createStream(
				ssc, auth);
		// ExtendedTwitterStream tweets = new
		// ExtendedTwitterStream(ssc.ssc());
		JavaDStream<String> statuses = tweets.filter(
				new Function<Status, Boolean>() {

					public Boolean call(Status v1) throws Exception {
						return true;
					}
				}).map(new Function<Status, String>() {

			public String call(Status v1) throws Exception {
				return v1.getText();
			}
		});
		tweets.print();
		ssc.start();
		ssc.awaitTermination();
	}
}